package com.yh.csx.bsf.ods.clickhouse;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.yh.csx.bsf.ods.BaseCanalConsumer.Operate;
import com.yh.csx.bsf.ods.BaseCanalConsumer.OperateEnum;
import com.yh.csx.bsf.ods.StragtegyProvider;
import com.yh.csx.bsf.ods.clickhouse.ClickhouseTable.ClickHouseColumn;
import com.yh.csx.bsf.ods.meta.MysqlDataType;
import com.yh.csx.bsf.ods.meta.OdsColumn;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.meta.OdsTable;
import com.yh.csx.bsf.ods.utils.GlobalIDUtils;
import com.yh.csx.bsf.ods.utils.OdsTypeUtils;
import com.yh.csx.bsf.ods.meta.OdsColumn.OdsColumnBuilder;
import com.yh.csx.bsf.ods.meta.OdsIndex;

import lombok.extern.slf4j.Slf4j;
import ru.yandex.clickhouse.domain.ClickHouseDataType;

/**
 * @author Robin.Wang
 * @Date 2020-05-11
 */
@Slf4j
@Service
@ConditionalOnProperty(name = "bsf.clickhouse.enabled", havingValue = "true")
public class ClickhouseAdapter implements OdsSyncAdapter {

	@Autowired
	@Qualifier(value = "clickHouseDataSource")
	DataSource dataSource;
	@Autowired
	StragtegyProvider stragtegyProvider;

	/**
	 * 判断库是否存在
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createDatabase(String database) throws Exception {
		Assert.hasText(database, "table can't be null");
		String sql = "create database if not exists " + database;
		return executeQuery(sql);
	}
	/**
	 * 查询数据库列表
	 * */
	public List<String> showDatabases(){
		List<String> result=new ArrayList<String>();
		String sql = "show databases";
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();
			while (ret.next()) {
				result.add(ret.getString("name"));
			}
		} catch (Exception e) {
			log.error("Exception {}", e);
		}
		return result;
	}
	/**
	 * 查询数据库表列表
	 * */
	public List<String> showTables(String databases){
		List<String> result=new ArrayList<String>();
		String sql = "show tables from "+ databases;
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();
			while (ret.next()) {
				result.add(ret.getString("name"));
			}
		} catch (Exception e) {
			log.error("Exception {}", e);
		}
		return result;
	}
	@Override
	public boolean createViews() throws Exception {
		List<String> dbs=this.showDatabases();
		for(String db:dbs) {
			List<String> tables=this.showTables(db);
			for(String table:tables) {
				Map<String,String> columns=this.describeTable(db+"."+table);
				
				StringBuilder sb=new StringBuilder();
				sb.append("create view if not exists ").append(db+"."+table+"_view");
				sb.append(" as select ");
				for(String column:columns.keySet()) {
					sb.append("argMax(").append(column).append(",").append("auto_id").append(")").append(",");
				}
				sb.deleteCharAt(sb.length()-1);
				sb.append("from ").append(db+"."+table).append(" group by id");
				this.executeQuery(sb.toString());
			}
		}
		return true;
	}
	
	/**
	 * 判断表是否存在
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean existsTable(String table) throws Exception {
		Assert.hasText(table, "table can't be null");
		String sql = "exists " + table;
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();

			if (ret.next()) {
				Boolean b = ret.getBoolean("result");
				return b;
			}
			return false;
		} catch (Exception e) {
			log.error("Exception {}", e);
			throw e;
		}

	}

	/**
	 * 创建数据表
	 * 
	 * @param table        库名.表名
	 * @param List<Column> 数据列
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createTable(String table, List<OdsColumn> columns) throws Exception {
		Set<String> ids = new HashSet<String>();
		for (OdsColumn c : columns) {
			if ("PRI".equals(c.getColumnKey())) {
				ids.add(c.getColumnName());
			}
		}
		return createTable(table, columns, ids, 8192);
	}

	/**
	 * 创建数据表
	 * 
	 * @param schema       数据库名
	 * @param table        表名
	 * @param List<Column> 数据列
	 * @param Set<String>  主键
	 * @param granularity  索引粒度
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createTable(String table, List<OdsColumn> columns, Set<String> ids, Integer granularity)
			throws Exception {
		Assert.notEmpty(columns, "columns can't be emptry");
		StringBuilder sb = new StringBuilder();
		sb.append("create table if not exists ").append(table).append("(");
		for (OdsColumn column : columns) {
			ClickHouseDataType type = ClickhouseTypeUtil
					.mysqlType2ClickhouseType(MysqlDataType.valueOf(column.getDataType()));
			sb.append(column.getColumnName()).append(" ");
			if (type.equals(ClickHouseDataType.Decimal)) {
				if (column.isNullable()) {
					sb.append("Nullable(").append(ClickHouseDataType.Decimal64.toString()).append("(").append(column.getNumericScale()).append("))");
				} else {					
					sb.append(ClickHouseDataType.Decimal64.toString()).append("(").append(column.getNumericScale()).append(")");
					
				}
			} else {
				if (column.isNullable()) {
					sb.append("Nullable(").append(type.toString()).append(")");
				} else {
					sb.append(type.toString());
				}
			}
			sb.append(",");
		}
		if (!sb.toString().contains("auto_id")) {
			sb.append("auto_id  Int64,");
		}
		if (!sb.toString().contains("auto_create_date")) {
			sb.append("auto_create_date Date default now(),");
		}
		sb.delete(sb.length() - 1, sb.length());// remove the last comma ","
		sb.append(")").append(" ENGINE = MergeTree(auto_create_date,");
		if (ids != null && ids.size() > 0) {
			sb.append("(").append(String.join(",", ids)).append("),");
		} else {
			sb.append("( auto_id ),");
		}
		sb.append(granularity > 0 ? granularity : 8192).append(");");
		return executeQuery(sb.toString());
	}

	public boolean createTable0(String table, List<OdsColumn> columns, Set<String> ids, Integer granularity)
			throws Exception {
		Assert.notEmpty(columns, "columns can't be emptry");
		StringBuilder sb = new StringBuilder();
		sb.append("create table ").append(table).append("(");
		for (OdsColumn column : columns) {
			sb.append(column.getColumnName()).append(" ");
			ClickHouseDataType type = ClickhouseTypeUtil
					.mysqlType2ClickhouseType(MysqlDataType.valueOf(column.getDataType().toUpperCase()));
			if (type.equals(ClickHouseDataType.Decimal)) {
				if (column.isNullable()) {
					sb.append("Nullable(").append(ClickHouseDataType.Decimal64.toString()).append("(").append(column.getNumericScale()).append("))");
				} else {
					sb.append(ClickHouseDataType.Decimal64.toString()).append("(").append(column.getNumericScale()).append(")");
				}
			} else {
				if (column.isNullable()) {
					sb.append("Nullable(").append(type.toString()).append(")");
				} else {
					sb.append(type.toString());
				}
			}
			sb.append(",");
		}
		if (!sb.toString().contains("auto_id")) {
			sb.append("auto_id  Int64,");
		}
		if (!sb.toString().contains("deleted")) {
			sb.append("deleted int,");
		}
		if (!sb.toString().contains("auto_create_date")) {
			sb.append("auto_create_date Date default now(),");
		}
		sb.delete(sb.length() - 1, sb.length());// remove the last comma ","
		sb.append(")").append(" ENGINE = MergeTree(auto_create_date,");
		if (ids != null && ids.size() > 0) {
			sb.append("(").append(String.join(",", ids)).append("),");
		} else {
			sb.append("( auto_id ),");
		}
		sb.append(granularity > 0 ? granularity : 8192).append(");");
		return executeQuery(sb.toString());
	}

	/**
	 * 如果库表不存在，则创建数据库
	 * 
	 * @param database 数据库
	 * @param table    数据表
	 * @param columns  数据列
	 * @param ids      主键字段名
	 */
	public boolean createTableIfNotExists(String database, String table, List<OdsColumn> columns, Set<String> ids)
			throws Exception {
		if (StringUtils.hasLength(database)) {
			this.createDatabase(database);
		}
		if (!existsTable(database + "." + table)) {
			return createTable(database + "." + table, columns, ids, 8192);
		}
		return true;
	}
	@Override
	public synchronized boolean createTableIfNotExists(String dbtable, List<OdsColumn> columns) throws Exception {
		String[] dbts=dbtable.split("\\.");
		return createTableIfNotExists(dbts[0], dbts[1],  columns);
		
	}
	@Override
	public boolean createTableIfNotExists(String database, String table, List<OdsColumn> columns) throws Exception {
		if (StringUtils.hasLength(database)) {
			this.createDatabase(database);
		}
		Set<String> ids = new HashSet<String>();
		for (OdsColumn column : columns) {
			if ("PRI".equals(column.getColumnKey())) {
				ids.add(column.getColumnName());
			}
		}
		if (ids.isEmpty()) {
			throw new RuntimeException(database + "." + table + "不存在主键");
		}
		if (!existsTable(database + "." + table)) {
			return createTable0(database + "." + table, columns, ids, 8192);
		}
		return true;
	}

	/**
	 * 查询表字段
	 * 
	 * @param table 库名.表名
	 * @return Map<String,String> 表列
	 */
	public Map<String, String> describeTable(String table) {
		Map<String, String> columns = new HashMap<String, String>();
		String sql = "describe " + table;
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();
			while (ret.next()) {
				columns.put(ret.getString("name"), ret.getString("type"));
			}
		} catch (Exception e) {
			log.error("Exception {}", e);
		}
		return columns;
	}

	/**
	 * 清空表
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 * 
	 */
	public boolean truncatTable(String table) throws Exception {
		String sql = "truncate table " + table;
		return executeQuery(sql);
	}

	/**
	 * 删除表
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 * 
	 */
	public boolean dropTable(String table) throws Exception {
		if (existsTable(table)) {
			String sql = "drop table if exists " + table;
			return executeQuery(sql);
		}
		return true;
	}

	/**
	 * 重命名表名
	 * 
	 * @param oldName 原表名
	 * @param newName 新表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean renameTable(String oldName, String newName) throws Exception {
		String sql = "rename table " + oldName + " to " + newName;
		return executeQuery(sql);
	}

	/**
	 * 增加列
	 * 
	 * @param table        库名.表名
	 * @param List<Column> 增加的表列
	 * @throws Exception
	 */
	public boolean alterTableAddColumn(String table, List<OdsColumn> columns) throws Exception {
		StringBuilder sb = new StringBuilder();
		sb.append("alter table ").append(table).append(" ");
		for (OdsColumn column : columns) {
			ClickHouseDataType type = ClickhouseTypeUtil
					.mysqlType2ClickhouseType(MysqlDataType.valueOf(column.getDataType().toUpperCase()));
			sb.append("add column ").append(column.getColumnName()).append(" ");
			if (type.equals(ClickHouseDataType.Decimal)) {
				sb.append(type.toString()).append("(").append(column.getNumericPrecision()).append(",")
						.append(column.getNumericScale()).append(")");
			} else {
				sb.append(type.toString());
			}
			sb.append(",");
		}

		Map<String, String> oldColumns = this.describeTable(table);
		if (!oldColumns.containsKey("auto_id")) {
			sb.append("add column auto_id Int64,");
		}
		if (!oldColumns.containsKey("auto_create_date")) {
			sb.append("add column auto_create_date Date default now(),");
		}
		sb.delete(sb.length() - 1, sb.length());// remove the last comma ","
		return executeQuery(sb.toString());
	}

	/**
	 * 插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 * @throws Exception
	 */
	public boolean insert(String table, List<OdsColumn> columns) throws Exception {
		StringBuilder sb = new StringBuilder();
		sb.append("insert into ").append(table).append("(");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(columns.get(i).getColumnName()).append(",");
		}
		sb.append("auto_id");
		sb.append(") ").append("values (");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(ClickhouseTypeUtil.perseValue(
					ClickhouseTypeUtil.mysqlType2ClickhouseType(
							MysqlDataType.valueOf(columns.get(i).getDataType().toUpperCase())),
					columns.get(i).getValue())).append(",");
		}
		sb.append(GlobalIDUtils.nextId());
		sb.append(")");
		return executeQuery(sb.toString());
	}

	/**
	 * 插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 */
	public String insertString(String table, List<OdsColumn> columns) {
		StringBuilder sb = new StringBuilder();
		sb.append("insert into ").append(table).append("(");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(columns.get(i).getColumnName()).append(",");
		}
		sb.append("auto_id");
		sb.append(") ").append("values (");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(columns.get(i).getValue()).append(",");
		}
		sb.append(GlobalIDUtils.nextId());
		sb.append(")");
		return sb.toString();
	}

	/**
	 * 批量插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 * @throws Exception
	 */
	public boolean insertBatch(String table, List<List<ClickHouseColumn>> data) throws Exception {
		StringBuilder sb = new StringBuilder();
		sb.append("insert into ").append(table).append(" (");
		for (int i = 0; i < data.get(0).size(); i++) {
			sb.append(data.get(0).get(i).getName()).append(",");
		}
		sb.append("auto_id");
		sb.append(") values ");
		for (int i = 0; i < data.size(); i++) {
			sb.append("(");
			for (int j = 0; i < data.get(i).size(); i++) {
				sb.append(ClickhouseTypeUtil.perseValue(data.get(i).get(j).getType(), data.get(i).get(j).getValue()))
						.append(",");
			}
			sb.append(GlobalIDUtils.nextId());
			sb.append("),");
		}
		sb.delete(sb.length() - 1, sb.length());
		return executeQuery(sb.toString());
	}

	/**
	 * 批量插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 * @throws Throwable 
	 */
	public boolean insertBatchMap(String table, List<Map<String, Object>> data, OdsTable tableStructure) throws Exception {
		StringBuilder sb = new StringBuilder();
		sb.append("insert into ").append(table).append(" (");
		for (OdsColumn column : tableStructure.getColumns()) {
			sb.append(column.getColumnName()).append(",");
		}
		boolean existsAutoid = sb.indexOf("auto_id") > 0;
		if (existsAutoid) {
			sb.delete(sb.length() - 1, sb.length());
		} else {
			sb.append("auto_id");
		}
		sb.append(") values ");
		for (int i = 0; i < data.size(); i++) {
			sb.append("(");
			for (OdsColumn column : tableStructure.getColumns()) {
				ClickHouseDataType type = ClickhouseTypeUtil
						.mysqlType2ClickhouseType(MysqlDataType.valueOf(column.getDataType().toUpperCase()));
				if (data.get(i).get(column.getColumnName()) != null) {
					sb.append(ClickhouseTypeUtil.perseValue(type, data.get(i).get(column.getColumnName()))).append(",");
				} else {
					if (type.equals(ClickHouseDataType.String)) {
						sb.append("'',");
					} else {
						sb.append(",");
					}
				}

			}
			if (existsAutoid) {

			} else {
				sb.append(GlobalIDUtils.nextId()).append(",");
			}
			sb.delete(sb.length() - 1, sb.length());
			sb.append("),");
		}
		sb.delete(sb.length() - 1, sb.length());
		try {
			return executeQuery(sb.toString());
		}catch(Exception e) {			
			this.syncStructure(table,tableStructure.getColumns());
			return  executeQuery(sb.toString());
		}
	}

	/**
	 * 替换
	 * 
	 * @throws Exception
	 */
	public boolean replace(String table, List<List<OdsColumn>> data) throws Exception {
		return executeQuery(replaceBatchString(table, data));
	}

	/**
	 * 替换
	 */
	private String replaceBatchString(String table, List<List<OdsColumn>> data) {
		StringBuilder sb = new StringBuilder();
		sb.append("replace into ").append(table).append(" (");
		for (int i = 0; i < data.get(0).size(); i++) {
			sb.append(data.get(0).get(i).getColumnName()).append(",");
		}
		sb.delete(sb.length() - 1, sb.length());
		sb.append(") values ");
		for (int i = 0; i < data.size(); i++) {
			sb.append("(");
			for (int j = 0; j < data.get(i).size(); j++) {
				sb.append(ClickhouseTypeUtil.perseValue(
						ClickhouseTypeUtil.mysqlType2ClickhouseType(
								MysqlDataType.valueOf(data.get(i).get(j).parseColumnType().getDataType())),
						data.get(i).get(j).getValue().toString())).append(",");
			}
			sb.delete(sb.length() - 1, sb.length());
			sb.append("),");
		}
		sb.delete(sb.length() - 1, sb.length());
		sb.append(")");
		return sb.toString();
	}

	public String buildSql(OperateEnum eventType, String table, List<OdsColumn> beforeColumns,
			List<OdsColumn> afterColumns) {
		switch(eventType) {
		case insert:
			return insertString(table, afterColumns);
		case update:
			return insertString(table, afterColumns);
		case delete:
			beforeColumns.add(new OdsColumn("deleted", false, "tinyint", "1"));// 自定义列
			return insertString(table, beforeColumns);
		}
		
		return "";
	}

	@Override
	public void syncStructure(OdsTable table) throws Exception {
		 this.syncStructure(table.getSchema(), table.getTableName(), table.getColumns());
	}
	private void syncStructure(String database,String table, List<OdsColumn> columns) throws Exception {
		this.createDatabase(database);
		this.createTableIfNotExists(database+"."+table, columns);
	}
	private void syncStructure(String databasetable, List<OdsColumn> columns) throws Exception {
		String[] dbt=databasetable.split("\\.");
		this.createDatabase(dbt[0]);
		this.createTableIfNotExists(databasetable, columns);
		Map<String, String> cols = this.describeTable(databasetable);
		List<OdsColumn> newColumns = new ArrayList<OdsColumn>();
		for (OdsColumn col : columns) {
			if (!cols.containsKey(col.getColumnName())) {
				newColumns.add(col);
			}
		}
		if (newColumns.size() > 0 || !cols.containsKey("auto_id")
				|| !cols.containsKey("auto_create_date")) {
			this.alterTableAddColumn(databasetable,	newColumns);
		}
	}
	
	private void syncStrutcureAndRetry(Header header, List<OdsColumn> afterColumns) throws Exception {
		try {

			if (!this.existsTable(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()))) {
				this.createDatabase(header.getSchemaName());
				this.createTable(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()),
						afterColumns);
			} else {
				Map<String, String> columns = this.describeTable(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()));
				List<OdsColumn> newColumns = new ArrayList<OdsColumn>();
				for (OdsColumn col : afterColumns) {
					if (!columns.containsKey(col.getColumnName())) {
						newColumns.add(col);
					}else {
						
					}
				}
				if (newColumns.size() > 0 || !columns.containsKey("auto_id")
						|| !columns.containsKey("auto_create_date")) {
					this.alterTableAddColumn(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()),
							newColumns);
				}
			}
			this.insert(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()),afterColumns);
		} catch (Exception e) {
			throw e;
		}

	}


	@Override
	public boolean process(List<Operate> list) throws Exception {
		StringBuilder sb = new StringBuilder();
		for (Operate o : list) {
			sb.append(this.buildSql(o.getOperateEnum(),
					stragtegyProvider.getTable(o.getHeader().getSchemaName(), o.getHeader().getTableName()),
					OdsTypeUtils.convert2OdsColumn(o.getBeforeColumns()),
					OdsTypeUtils.convert2OdsColumn(o.getAfterColumns()))).append(";");
		}
		return this.executeQuery(sb.toString());
	}

	@Override
	public void insert(Header header, List<Column> afterColumns) throws Exception {
		try {
			this.insert(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()),
					OdsTypeUtils.convert2OdsColumn(afterColumns));
		} catch (Exception e) {
			syncStrutcureAndRetry(header, OdsTypeUtils.convert2OdsColumn(afterColumns));
		}
	}

	@Override
	public void update(Header header, List<Column> beforeColumns, List<Column> afterColumns) throws Exception {
		try {
			this.insert(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()),
					OdsTypeUtils.convert2OdsColumn(afterColumns));
		} catch (Exception e) {
			syncStrutcureAndRetry(header, OdsTypeUtils.convert2OdsColumn(afterColumns));
		}
	}

	@Override
	public void delete(Header header, List<Column> beforeColumns) throws Exception {
		List<OdsColumn> columns = OdsTypeUtils.convert2OdsColumn(beforeColumns);
		columns.add(OdsColumnBuilder.build().setColumnName("deleted").setValue(1).setDataType("tinyint"));
		this.insert(stragtegyProvider.getTable(header.getSchemaName(), header.getTableName()), columns);
	}

	/**
	 * 执行查询
	 * 
	 * @throws Exception
	 */
	public boolean executeQuery(String sql) throws Exception {
		if (StringUtils.hasText(sql)) {
			try (Connection conn = dataSource.getConnection()) {
				PreparedStatement prepareStatement = conn.prepareStatement(sql);
				prepareStatement.executeQuery();
				return true;
			} catch (Exception e) {
				log.error("数据操作异常:{},{}",sql,e);
				throw e;
			}
		}
		return true;
	}
	@Override
	public boolean createIndex(String tableName, List<OdsIndex> indexs) throws Exception {
		// TODO Auto-generated method stub
		return false;
	}
	@Override
	public Integer selectCount(String sql) {
		// TODO Auto-generated method stub
		return null;
	}
	@Override
	public ResultSet query(String sql) throws Exception {
		// TODO Auto-generated method stub
		return null;
	}
	@Override
	public void insertBatch(String dataTable, Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		
	}
	@Override
	public void updateBatch(String dataTable, Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		
	}
	@Override
	public void deleteBatch(String dataTable, Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		
	}


}
